--------- MODULE downloader ------------

EXTENDS Integers, Naturals, Sequences, TLC, FiniteSets

Last(seq) == seq[Len(seq)]

Empty(seq) == Len(seq) = 0

RECURSIVE Take(_,_)
Take(n, seq) == CASE n = 0 -> <<>>
  [] n = 1 -> <<Head(seq)>>
  [] n >= 2 -> Take(1, seq) \o Take(n-1, Tail(seq))

RECURSIVE Drop(_,_)
Drop(n, seq) == CASE n = 0 -> seq
  [] n = 1 -> Tail(seq)
  [] n >= 2 -> Drop(n-1, Drop(1, seq))

GapsEqual(seq) ==
  LET
    Ixs == DOMAIN seq
    SubsequentPairs == {pair \in Ixs \X Ixs: pair[1] + 1 = pair[2]}
    AllGaps == {seq[pair[1]] - seq[pair[2]]: pair \in SubsequentPairs}
  IN Cardinality(AllGaps) <= 1

MergeTradeIntoBuckets(buckets, trade) == Append(buckets, trade)

(*--algorithm downloader
variables
  history = [ x \in 1 .. 10 |-> x ],
  all_trades = [ i \in 1 .. 10 |-> i + Last(history) ],
  bx_trades_left = all_trades,
  buckets = <<>>,
  trade_batch = <<>>,

define
  EventuallySynched == <>[](buckets = history \o all_trades)
  Invariant == GapsEqual(buckets)
end define;

procedure receive_batch()
  variable possible_batch_sizes = 1 .. Len(bx_trades_left);
begin
  ReceiveBatch:
    with batch_size \in possible_batch_sizes do
      trade_batch := Take(batch_size, bx_trades_left);
      bx_trades_left := Drop(batch_size, bx_trades_left);
      return;
    end with;
end procedure;

procedure merge_batch()
  variable trade_being_merged = "null";
begin
  MergeBatch2:
    while ~Empty(trade_batch) do
      trade_being_merged := Head(trade_batch);
      trade_batch := Tail(trade_batch);
      buckets := MergeTradeIntoBuckets(buckets, trade_being_merged);
    end while;
    return;
end procedure;

fair process p1 = "p1"
begin
  GetHistory:
    buckets := buckets \o history;
  UpdateBasedOnIncomingTrades:
    while ~Empty(bx_trades_left) do
      call receive_batch();
      MergeBatch:
        call merge_batch();
    end while;
end process;

end algorithm;*)
\* BEGIN TRANSLATION
VARIABLES bx_history, bx_trades, bx_trades_left, buckets, trade_batch, 
          trade_being_merged, pc, stack

(* define statement *)
EventuallySynched == <>[](Synched(bx_history,bx_trades,buckets))
Invariant == GapsEqual(buckets)

VARIABLE possible_batch_sizes

vars == << bx_history, bx_trades, bx_trades_left, buckets, trade_batch, 
           trade_being_merged, pc, stack, possible_batch_sizes >>

ProcSet == {"p1"}

Init == (* Global variables *)
        /\ bx_history = [ x \in 1 .. 10 |-> x ]
        /\ bx_trades = [ i \in 1 .. 10 |-> i + Last(bx_history) ]
        /\ bx_trades_left = bx_trades
        /\ buckets = <<>>
        /\ trade_batch = <<>>
        /\ trade_being_merged = 0
        (* Procedure ReceiveBatch *)
        /\ possible_batch_sizes = [ self \in ProcSet |-> 1 .. Len(bx_trades_left)]
        /\ stack = [self \in ProcSet |-> << >>]
        /\ pc = [self \in ProcSet |-> "WaitForHistory"]

FeedTrades(self) == /\ pc[self] = "FeedTrades"
                    /\ \E batch_size \in possible_batch_sizes[self]:
                         /\ trade_batch' = Take(batch_size, bx_trades_left)
                         /\ bx_trades_left' = Drop(batch_size, bx_trades_left)
                         /\ pc' = [pc EXCEPT ![self] = Head(stack[self]).pc]
                         /\ possible_batch_sizes' = [possible_batch_sizes EXCEPT ![self] = Head(stack[self]).possible_batch_sizes]
                         /\ stack' = [stack EXCEPT ![self] = Tail(stack[self])]
                    /\ UNCHANGED << bx_history, bx_trades, buckets, 
                                    trade_being_merged >>

ReceiveBatch(self) == FeedTrades(self)

WaitForHistory == /\ pc["p1"] = "WaitForHistory"
                  /\ buckets' = buckets \o bx_history
                  /\ pc' = [pc EXCEPT !["p1"] = "UpdateBasedOnIncomingTrades"]
                  /\ UNCHANGED << bx_history, bx_trades, bx_trades_left, 
                                  trade_batch, trade_being_merged, stack, 
                                  possible_batch_sizes >>

UpdateBasedOnIncomingTrades == /\ pc["p1"] = "UpdateBasedOnIncomingTrades"
                               /\ IF ~Empty(bx_trades_left)
                                     THEN /\ stack' = [stack EXCEPT !["p1"] = << [ procedure |->  "ReceiveBatch",
                                                                                   pc        |->  "MergeBatch",
                                                                                   possible_batch_sizes |->  possible_batch_sizes["p1"] ] >>
                                                                               \o stack["p1"]]
                                          /\ possible_batch_sizes' = [possible_batch_sizes EXCEPT !["p1"] = 1 .. Len(bx_trades_left)]
                                          /\ pc' = [pc EXCEPT !["p1"] = "FeedTrades"]
                                     ELSE /\ pc' = [pc EXCEPT !["p1"] = "Done"]
                                          /\ UNCHANGED << stack, 
                                                          possible_batch_sizes >>
                               /\ UNCHANGED << bx_history, bx_trades, 
                                               bx_trades_left, buckets, 
                                               trade_batch, trade_being_merged >>

MergeBatch == /\ pc["p1"] = "MergeBatch"
              /\ IF ~Empty(trade_batch)
                    THEN /\ trade_being_merged' = Head(trade_batch)
                         /\ trade_batch' = Tail(trade_batch)
                         /\ buckets' = MergeTradeIntoBuckets(buckets, trade_being_merged')
                         /\ pc' = [pc EXCEPT !["p1"] = "MergeBatch"]
                    ELSE /\ pc' = [pc EXCEPT !["p1"] = "UpdateBasedOnIncomingTrades"]
                         /\ UNCHANGED << buckets, trade_batch, 
                                         trade_being_merged >>
              /\ UNCHANGED << bx_history, bx_trades, bx_trades_left, stack, 
                              possible_batch_sizes >>

p1 == WaitForHistory \/ UpdateBasedOnIncomingTrades \/ MergeBatch

Next == p1
           \/ (\E self \in ProcSet: ReceiveBatch(self))
           \/ (* Disjunct to prevent deadlock on termination *)
              ((\A self \in ProcSet: pc[self] = "Done") /\ UNCHANGED vars)

Spec == /\ Init /\ [][Next]_vars
        /\ WF_vars(p1) /\ WF_vars(ReceiveBatch("p1"))

Termination == <>(\A self \in ProcSet: pc[self] = "Done")

\* END TRANSLATION

===========

Noriu parasyti spec'a tam ka rasau piesejui
  - kainos istorijos ir kintancios dabarties parsiuntimas

bendras algoritmas
  - parsiusti kas buvo praeityje bucket'ais
  - pakartotinai
    - klausytis pakitimu informacijos
    - apdoroti ja i einama bucketa
      - kuris iki savo gyvenimo pabaigos reaguoja i nauja informacija
    - jeigu einamas bucketas pasiekia pabaiga
      - prideti ji prie praeities bucketu
      - ir pereiti prie naujo

reikalavimai
  - kad neatsirastu tarpu
  - kad nebutu prarasta informacija
